package com.tmsb.sink;

import com.tmsb.pojo.FairShopOrderVerifiedNetwork;
import com.tmsb.utils.ConfUtil;
import com.tmsb.utils.RedisCluster;
import com.tmsb.utils.Util;
import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.api.common.typeinfo.TypeHint;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.runtime.state.FunctionInitializationContext;
import org.apache.flink.runtime.state.FunctionSnapshotContext;
import org.apache.flink.streaming.api.checkpoint.CheckpointedFunction;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import redis.clients.jedis.JedisCluster;

import java.text.SimpleDateFormat;
import java.util.Date;

/**
 * Description: 用tmsb_fair_shop的order_verified_network来获取每日水核销数指标的sink
 * Version:1.0.0
 *
 * @author wxw
 * @date 2020/7/27
 */
public class OrderVerifiedNetworkSink extends RichSinkFunction<FairShopOrderVerifiedNetwork> implements CheckpointedFunction {
    //TODO 总核销水数量
    private ListState<Long> totalVerifyQtyState;
    private Long totalVerifyQty;
    private ListState<Long> dateState;
    private Long dateValue;
    private SimpleDateFormat sdf;
    private JedisCluster jedis;

    @Override
    public void snapshotState(FunctionSnapshotContext context) throws Exception {
        totalVerifyQtyState.clear();
        dateState.clear();
        totalVerifyQtyState.add(totalVerifyQty);
        dateState.add(dateValue);
    }

    @Override
    public void initializeState(FunctionInitializationContext context) throws Exception {
        totalVerifyQty = 0L;
        dateValue = 0L;
        jedis = RedisCluster.getInstance(ConfUtil.getRedisCluster()).getRedisClusterList().get(0);
        sdf = new SimpleDateFormat("yyyyMMdd");
        ListStateDescriptor<Long> totalVerifyQtyStateDes =
                new ListStateDescriptor<>("totalVerifyQtyState", TypeInformation.of(new TypeHint<Long>() {}));
        ListStateDescriptor<Long> ovn_dateDes =
                new ListStateDescriptor<>("ovn_date", TypeInformation.of(new TypeHint<Long>() {}));
        totalVerifyQtyState = context.getOperatorStateStore().getListState(totalVerifyQtyStateDes);
        dateState = context.getOperatorStateStore().getListState(ovn_dateDes);
        if (context.isRestored()) {
            totalVerifyQty = totalVerifyQtyState.get().iterator().next();
            dateValue = dateState.get().iterator().next();
        }
    }

    @Override
    public void invoke(FairShopOrderVerifiedNetwork value, Context context) throws Exception {
        //TODO 每天凌晨清空状态
         long create_time = value.getCreate_time()-28800000;
         if(create_time >= dateValue){
             totalVerifyQty = 0L;
             totalVerifyQtyState.clear();
             dateValue = Util.getTomorrowZeroPointTimestamps();
         }else if(create_time < Util.getZeroPointTimestamps()){
             return;
         }
        //TODO 其余的就是今天的订单，计算每日核销水数量
         long verify_qty = value.getVerify_qty();
         totalVerifyQty += verify_qty;
         //TODO 向redis发送数据
        String format = sdf.format(new Date(create_time));
        jedis.hset(ConfUtil.getOrderVerifyQtyRedisKey(),format,totalVerifyQty+"");

    }

    @Override
    public void close() throws Exception {
        super.close();
        jedis.close();
    }
}
